package com.zt.bigdata.practise.javastrategy.chapter9;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.time.Instant;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;

/**
 *
 */
public class Chapter9 {
    private Map<Integer, Product> cache = new HashMap<>();
    private Logger logger = LoggerFactory.getLogger(this.getClass().getName());

    public static void main(String[] args) {
        //9.1 将顺序流转换为并行流
        System.out.println(Stream.of(1, 2, 3, 4, 5).isParallel());
        System.out.println(Stream.iterate(1, n -> n + 1).isParallel());
        System.out.println(Stream.generate(Math::random).limit(10).isParallel());

        //并行流串行流转换
        System.out.println(Arrays.asList(3, 1, 4, 1, 5, 9).parallelStream().isParallel());
        System.out.println(Arrays.asList(3, 1, 4, 1, 5, 9).parallelStream().sequential().isParallel());

        List<Integer> numbers = Arrays.asList(3, 1, 4, 1, 5, 9);
        // 流在达到终止表达式(terminal expression)前不会进行任何操作，
        // 即在达到终止表达式时才会评估流的状态。
        // 在本例中， 由于 collect 方法之前最后调用的是 sequential 方法，
        // 程序将返回顺序流，并对元素做相 应处理。
        long start = System.currentTimeMillis();
        List<Integer> nums = numbers.parallelStream()
                .map(n -> n * 2)
                .peek(n -> System.out.printf("%s processing %d%n", Thread.currentThread().getName(), n))
                .sequential().sorted()
                .collect(Collectors.toList());
        System.out.println(System.currentTimeMillis() - start);
        start = System.currentTimeMillis();
        numbers.parallelStream()
                .peek(n -> System.out.printf("%s processing %d%n", Thread.currentThread().getName(), n))
                .sorted()
                .collect(Collectors.toList());
        System.out.println(System.currentTimeMillis() - start);


        Instant before = Instant.now();
        int total = IntStream.of(3, 1, 4, 1, 5, 9)
                .map(Chapter9::doubleIt)
                .sum();
        Instant after = Instant.now();
        Duration duration = Duration.between(before, after);
        System.out.println("Total of doubles = " + total);
        System.out.println("time = " + duration.toMillis() + " ms");

        before = Instant.now();
        total = IntStream.of(3, 1, 4, 1, 5, 9).parallel()
                .map(Chapter9::doubleIt).sum();
        after = Instant.now();
        duration = Duration.between(before, after);
        System.out.println("Total of doubles = " + total);
        System.out.println("time = " + duration.toMillis() + " ms");

        //9.4 Future接口
        //用户希望执行表示异步计算结果、检查计算是否完成、
        // 在必要时取消计算、检索计算结果 等操作。

        ExecutorService service = Executors.newCachedThreadPool();
        Future<String> future = service.submit(() -> {
            Thread.sleep(100);
            return "hello world";
        });
        System.out.println("Processing");

        while (!future.isDone()) {
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("Waiting...");
        }

        if (!future.isCancelled()) {
            try {
                System.out.println(future.get()); //阻塞操作
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            } finally {
            }
        } else {
            System.out.println("Cancelled");
        }

        future = service.submit(() -> {
            Thread.sleep(10);
            return "Hello, World!";
        });
        future.cancel(true);
        System.out.println("Even more processing...");
        service.shutdown();

//        future.get()
        //显示通知程序执行完成
        //9.5 完成CompletableFuture
        Chapter9 demo = new Chapter9();
        try {
//            demo.getProduct(666).get();
            demo.getProduct(1);
        } catch (Exception e) {
            e.printStackTrace();
        }

        demo.getProductAsync(2);


        //9.6 多个CompletableFuture之间的协调(第1部分)
        CompletableFuture.supplyAsync(Chapter9::sleepThenReturnString)
                .thenApply(Integer::parseInt)
                .thenApply(x -> x * 2)
                .thenAccept(System.out::println)
                .join();//join 方法是阻塞方法
        System.out.println("Running ...");


        ExecutorService executorService = Executors.newFixedThreadPool(4);
        CompletableFuture.supplyAsync(Chapter9::sleepThenReturnString, executorService)
                .thenApply(Integer::parseInt)
                .thenApply(x -> 2 * x)
                .thenAccept(System.out::println)
                .join();
        System.out.println("Running...");

        int x = 2;
        int y = 3;
        //thenCompose 方法的参数是一个函数，
        // 它传入第一个 Future 的结果，并将其转换为第二个 Future 的输出
        //两个 Future 的复合
        CompletableFuture<Integer> completableFuture =
                CompletableFuture.supplyAsync(() -> x)
                        .thenCompose(n -> CompletableFuture.supplyAsync(() -> n + y));
        System.out.println(completableFuture.join());

        //两个 Future 的合并
        CompletableFuture<Integer> completableFuture1 =
                CompletableFuture.supplyAsync(() -> x)
                        .thenCombine(
                                CompletableFuture.supplyAsync(() -> y),
                                (n1, n2) -> n1 + n2
                        );
        System.out.println(completableFuture1.join());

        String num = "abc";
        CompletableFuture<Integer> value = getIntegerCompletableFuture(num);
        System.out.println(value.join());

        num = "42";
        value = getIntegerCompletableFuture(num);
        System.out.println(value.join());


        try {
            CompletableFuture<Product> future1 = demo.getProduct(1);
            //1
            System.out.println("使用join: " + future1.join());
            //2
            System.out.println("使用get: " + future1.get());

            Supplier<CompletableFuture<Product>> excepion = () -> {
                System.out.println("发生异常");
                CompletableFuture<Product> f = new CompletableFuture<>();
                f.completeExceptionally(new RuntimeException(""));
                return f;
            };
            //3
            System.out.println("异常处理: " + excepion.get().exceptionally(throwable -> new Product(1, "33")).join());

            future1.thenAcceptAsync(product -> System.out.println("使用thenAcceptAsync: " + product));
            System.out.println("使用handle: " +
                    excepion.get().handle(((product, throwable) -> {
                        if (throwable != null) {
                            return "异常";
                        } else {
                            return product.toString();
                        }
                    })).join()
            );
            System.out.println("使用后继apply: " +
                    future1.thenApply(product -> "产品:" + product).get()
            );

            future1.whenComplete(((product, throwable) -> {
                if (throwable != null) {
                    System.out.println("异常");
                } else {
                    System.out.println("使用 whenComplete 产品:" + product);
                }
            }));


            CompletableFuture<Product> f = new CompletableFuture<>();
            f.complete(new Product(1, "222"));
            System.out.println("使用 complete 产品:" + f.join());

            CompletableFuture.runAsync(() -> System.out.println("异步执行 runAsync "));


        } catch (Exception e) {
            e.printStackTrace();
        }

        executorService.shutdown();

    }

    private static CompletableFuture<Integer> getIntegerCompletableFuture(String num) {
        return CompletableFuture.supplyAsync(() -> Integer.parseInt(num))
                .handle(((integer, throwable) -> integer != null ? integer : 0));
    }

    private static String sleepThenReturnString() {
        try {
            Thread.sleep(100);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return "42";
    }

    public static int doubleIt(int n) {
        try {
            Thread.sleep(100);
        } catch (InterruptedException ignore) {
        }
        return n * 2;
    }

    private Product getLocal(int id) {
        return cache.get(id);
    }

    private Product getRemote(int id) {
        try {
            Thread.sleep(100);//模拟耗时操作
            if (id == 666) {
                throw new RuntimeException("Evil request");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return new Product(id, "name");
    }

    private CompletableFuture<Product> getProduct(int id) {
        try {
            Product product = getLocal(id);
            if (product != null) {
                return CompletableFuture.completedFuture(product);
            } else {
                CompletableFuture<Product> future = new CompletableFuture<>();
                Product p = getRemote(id);
                cache.put(id, p);
                future.complete(p);
                return future;
            }
        } catch (Exception e) {
            e.printStackTrace();
            CompletableFuture<Product> future = new CompletableFuture<>();
            future.completeExceptionally(e);
            return future;
        }
    }

    private CompletableFuture<Product> getProductAsync(int id) {
        try {
            Product product = getLocal(id);
            if (product != null) {
                logger.info("getLocal with id=" + id);
                return CompletableFuture.completedFuture(product);
            } else {
                logger.info("getRemote with id=" + id);
                return CompletableFuture.supplyAsync(() -> {
                    Product p = getRemote(id);
                    cache.put(id, p);
                    return p;
                });
            }
        } catch (Exception e) {
            logger.info("exception thrown");
            CompletableFuture<Product> future = new CompletableFuture<>();
            future.completeExceptionally(e);
            return future;
        }
    }

    static class Product {
        int id;
        String name;

        public Product(int id, String name) {
            this.id = id;
            this.name = name;
        }

        public int getId() {
            return id;
        }

        public void setId(int id) {
            this.id = id;
        }

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        @Override
        public String toString() {
            return "Product{" +
                    "id=" + id +
                    ", name='" + name + '\'' +
                    '}';
        }
    }
}
